package jp.aist.gtrc.plus.admission;

import java.rmi.RemoteException;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Collection;
import java.util.Date;
import java.util.HashSet;
import java.util.List;
import java.util.regex.MatchResult;
import java.util.Scanner;

//import jp.aist.gtrc.plus.admission.EDF.EDFAdmission;
import jp.aist.gtrc.plus.admission.EDF.EDFAdmissionUniform;
//import jp.aist.gtrc.plus.admission.EDF.CPU;
import jp.aist.gtrc.plus.admission.EDF.Job;
import jp.aist.gtrc.plus.admission.EDF.JobArray;
import jp.aist.gtrc.plus.admission.EDF.Schedule;
import jp.aist.gtrc.plus.admission.EDF.UniformCPU;

import jp.aist.gtrc.plus.scheduler.SchedulerOptions;

import jp.aist.gtrc.plus.scheduler.status.JobStatus;
import jp.aist.gtrc.plus.scheduler.status.MainServer;
import jp.aist.gtrc.plus.scheduler.status.NodeStatus;
import jp.aist.gtrc.plus.scheduler.status.StatusManager;

import jp.aist.gtrc.plus.scheduler.specific.pbs.status.PBSJobStatus;
import jp.aist.gtrc.plus.scheduler.specific.pbs.status.PBSNodeStatus;

import jp.aist.gtrc.plus.scheduler.sys.SchedulerException;

import org.apache.log4j.Logger;

public class AdmissionOperationManager implements AdmissionServer {

        public final static int CLEANUP_TIME = 30;   // time for pbs_server to clean up terminated job in seconds;
        public final static int MAX_CONFIRMATION_TIME = 600;   // time to wait for user confirmation since submission in seconds;

   protected static final  Logger logger = Logger.getLogger(AdmissionOperationManager.class);

        private StatusManager statusMgr;
        private MainServer mainServer;
        private boolean dirty_status_flag = true;;

        private Schedule schedule = null;
        private List<Job> temporaryJobs;
 
   public AdmissionOperationManager(StatusManager statusMgr, MainServer srv) throws AdmissionException {
           this.statusMgr = statusMgr;
           this.mainServer = srv;
   }

   /*
    * Test only if the jobArray can fit. No change in the current schedule
    */
   public synchronized boolean isFeasible(JobArray jobArray) throws AdmissionException,SchedulerException {
           boolean isOK = false;
           if (dirty_status_flag == true) {
              long before = System.currentTimeMillis();
              System.out.println("isFeasible() is about to update status");
              statusMgr.updateStatus(mainServer);
              clearStatusDirtyFlag();
              long after = System.currentTimeMillis();
              System.out.println("isFeasible() spent " + (after-before) + " msec. in statusMgr.updateStatus()");
           }
           Collection<JobStatus> qosJobs = getQoSJobs(statusMgr.getJobs());
           //for (JobStatus jobStatus : qosJobs) {
           //   System.out.println("QoS Job " + jobStatus.getName()); 
           //}
           Collection<UniformCPU> cpus = getAvailableCPUs(statusMgr.getNodes());
           if (cpus.size() == 0) {
              System.err.println("No render CPU available"); 
//              throw new AdmissionException("No render CPU available"); 
           }
           for (UniformCPU cpu : cpus) {
              System.out.println("CPU " + cpu.id + ", " + cpu.lastOccupied + ", " + cpu.performanceIndex + ", " + cpu.node);
           }
           Collection<Job> jobs = new ArrayList<Job>();

           /*
            * We internally rerun job id as an integer for easy comparison.
            */
           int id = 0; 
           Calendar currentTime = Calendar.getInstance();
           int now = (int) (currentTime.getTimeInMillis() / 1000);
           for (JobStatus jobStatus : qosJobs) {
              PBSJobStatus pbsJobStatus = (PBSJobStatus) jobStatus;
              int deadline = pbsJobStatus.getDeadline();
              if (deadline <= now) {
                 System.err.println("Deadline of job " + jobStatus.getJobID() + " already passed");
                 logger.error("Deadline of job " + jobStatus.getJobID() + " already passed");
              }
              int submitTime = (int) (pbsJobStatus.getSubmitTime().getTimeInMillis()/1000);
              Job job = new Job(id,jobStatus.getJobID().toString(),pbsJobStatus.getEstimatedExecutionTime()+CLEANUP_TIME,deadline,submitTime + MAX_CONFIRMATION_TIME);
              jobs.add(job);
              id++;
           }

           EDFAdmissionUniform admissionControl = new EDFAdmissionUniform();
           jobArray.exeTime = jobArray.exeTime + CLEANUP_TIME;
           isOK = admissionControl.buildSchedule(cpus,jobs,jobArray);
           this.schedule = admissionControl.getSchedule(); 
           this.temporaryJobs = admissionControl.getTemporaryJobs();
           return isOK;
}

        public synchronized void submit(JobArray jobArray,QcommandClient qcommandClient) throws AdmissionException, RemoteException, SchedulerException {

           if (isFeasible(jobArray)) {
              String jobId = qcommandClient.qcommand();            
              if (jobId != null) {  // do we need to check more for the correct job id?
                 changeTemporaryJobNames(jobId,jobArray);
              }
              setStatusDirtyFlag();
           } 
        }

   /* 
    * Rename temporary jobs
    */ 
   private void changeTemporaryJobNames(String jobId, JobArray jobArray) {
      Scanner sc = new Scanner(jobId);
      sc.findInLine("^(\\d+)\\[\\]\\.(.+)$"); // Torque new version; with []
      MatchResult result = sc.match();
      if (result.groupCount() == 2) {
         String jobArrayId = result.group(1);
         String host = result.group(2);
         int num = jobArray.firstJobNum;
         for (Job job : temporaryJobs) {
            job.name = jobArrayId + "[" + num + "]." + host;
            num++; 
         }
         return;
      }

      sc = new Scanner(jobId);
      sc.findInLine("^(\\d+)\\.(.+)$"); // Torque old version; no [] 
      result = sc.match();
      if (result.groupCount() == 2) {
         String jobArrayId = result.group(1);
         String host = result.group(2);
         int num = jobArray.firstJobNum;
         for (Job job : temporaryJobs) {
            job.name = jobArrayId + "-" + num + "." + host;
            num++; 
         }
         return;
      }

      throw new RuntimeException("Unexected job id pattern found: " + jobId);

   }
 
   /*
    * 
    */
   public synchronized int isFeasibleAndRecommend(JobArray jobArray) throws AdmissionException,SchedulerException {
           long before = System.currentTimeMillis();
           long after = before;
           if (dirty_status_flag == true) {
              System.out.println("isFeasibleAndRecommend() is about to update status");
              statusMgr.updateStatus(mainServer);
              clearStatusDirtyFlag();
              after = System.currentTimeMillis();
              //System.out.println("isFeasibleAndRecommend() spent " + (after-before) + " msec. in statusMgr.updateStatus()");
           }

           before = System.currentTimeMillis();
           Collection<JobStatus> qosJobs = getQoSJobs(statusMgr.getJobs());
           Collection<UniformCPU> cpus = getAvailableCPUs(statusMgr.getNodes());
           if (cpus.size() == 0) {
              throw new AdmissionException("No available CPU"); 
           }
           for (UniformCPU cpu : cpus) {
              System.out.println("CPU " + cpu.id + ", " + cpu.lastOccupied + ", " + cpu.performanceIndex + ", " + cpu.node);
           }
           Collection<Job> jobs = new ArrayList<Job>();

           /*
            * We internally rerun job id as an integer for easy comparison.
            */
           int id = 0; 
           Calendar currentTime = Calendar.getInstance();
           int now = (int) (currentTime.getTimeInMillis() / 1000);
           for (JobStatus jobStatus : qosJobs) {
              PBSJobStatus pbsJobStatus = (PBSJobStatus) jobStatus; 
              if (pbsJobStatus.isRunning()) continue;
              int deadline = pbsJobStatus.getDeadline();
              if (deadline <= now) {
                 System.err.println("Deadline of job " + jobStatus.getJobID() + " already passed");
                 logger.error("Deadline of job " + jobStatus.getJobID() + " already passed");
              }
              int submitTime = (int) (pbsJobStatus.getSubmitTime().getTimeInMillis()/1000);
              Job job = new Job(id,jobStatus.getJobID().toString(),pbsJobStatus.getEstimatedExecutionTime()+CLEANUP_TIME,deadline,submitTime + MAX_CONFIRMATION_TIME);
              jobs.add(job);
              id++;
           }

           EDFAdmissionUniform admissionControl = new EDFAdmissionUniform();
           jobArray.exeTime = jobArray.exeTime + CLEANUP_TIME;
           int newDeadline = admissionControl.buildScheduleAndRecommend(cpus,jobs,jobArray);
           this.schedule = admissionControl.getSchedule();
           this.temporaryJobs = admissionControl.getTemporaryJobs();
           after = System.currentTimeMillis();
           System.out.println("buildScheduleAndRecommend() spent " + (after-before) + " msec.");
           return newDeadline;
}

        public synchronized int submitAndRecommend(JobArray jobArray,QcommandClient qcommandClient) throws AdmissionException, RemoteException, SchedulerException {

           int newDeadline = 0;
           newDeadline = isFeasibleAndRecommend(jobArray);
           int confirmationDeadline = 0;
           Calendar calendar = Calendar.getInstance();
           if (newDeadline == 0) { // in case the deadline is OK 
              calendar.setTimeInMillis(((long) jobArray.deadline)*1000);
              confirmationDeadline = jobArray.deadline;
           } else { // requested deadline cannot be met
              calendar.setTimeInMillis(((long) newDeadline)*1000);
              confirmationDeadline = newDeadline;
           }
           SimpleDateFormat dateFormater = new SimpleDateFormat("yyyyMMddHHmm.ss");
           String posixDateTime = dateFormater.format(calendar.getTime());

           //System.out.println("Confirmation deadline of job " + jobArray.name + " = " + posixDateTime + "(" + confirmationDeadline + ")" );

           Job firstJob = this.temporaryJobs.get(0);

           int expectedStartTime = this.schedule.getExpectedStartTime(firstJob);
           //System.out.println(firstJob.name + " has exp time " + (new Date( ((long) expectedStartTime*1000))));
           int now = (int) (Calendar.getInstance().getTimeInMillis()/1000);
           int latestConfirmationTime = Math.min(expectedStartTime,now + MAX_CONFIRMATION_TIME);

           //String jobId = qcommandClient.qcommand();            
           String jobId = qcommandClient.qcommand("" + confirmationDeadline); // with the confirmation deadline            
//System.err.println("Put job " + jobId + " into PBS");
           if (jobId != null) {  // do we need to check more for the correct job id?
              changeTemporaryJobNames(jobId,jobArray);
//System.err.println("change job name " + jobId + " in schedule");
           }
           setStatusDirtyFlag();
   
           return newDeadline;
        }

        /*
         * Reschedule due to change by queue manipulation commands such as qdel
         */
        public synchronized void reschedule(QcommandClient qcommandClient) throws AdmissionException, SchedulerException, RemoteException {
           if (qcommandClient != null) {
              qcommandClient.qcommand();     // qcommandClient here is qdel instead
              try {
                 Thread.sleep(3000);    // wait a bit to make sure PBS queue is updated
              } catch (Exception e) {

              }
              setStatusDirtyFlag();
           }
           JobArray dummyJobArray = new JobArray("DummyJobArray",0,0,Integer.MAX_VALUE);
           if (isFeasible(dummyJobArray)) {
              System.err.println("AdmissionOperationManger reschedule() OK");
           } else {
              System.err.println("AdmissionOperationManger reschedule() fails!!! what to do?");
           }
        }

        /*
         * Return only QoS jobs in held and queued jobs (excluding
         *  running and complete jobs).
         */
        private Collection<JobStatus> getQoSJobs(Collection<JobStatus> jobs) {
                Collection<JobStatus> qosJobs = new ArrayList<JobStatus>();
                for (JobStatus jobStatus : jobs) {
                   PBSJobStatus pbsJobStatus = (PBSJobStatus) jobStatus;
                   if ( pbsJobStatus.isQoSJob() && (pbsJobStatus.isHeld() || pbsJobStatus.isQueued()) )  {
                      qosJobs.add(jobStatus);
                   }
                }
                return qosJobs;
        }

        /*
         * Return a set of uniform CPUs (a set of processor cores) with its 
         *  available time (lastOccupied). All nodes with property 
         *  "QoS" are returned.
         */
        private Collection<UniformCPU> getAvailableCPUs(Collection<NodeStatus> nodes) {
           Collection<UniformCPU> cpus = new ArrayList<UniformCPU>();
           for (NodeStatus nodeStatus : nodes) {
              int cpuCount = 0;
              PBSNodeStatus pbsNodeStatus = (PBSNodeStatus) nodeStatus;       
              if ( pbsNodeStatus.getAttribute("QoS") == null) {
                 continue;
              }  

              //System.out.println(pbsNodeStatus.getName());
              if (!pbsNodeStatus.isAlive()) continue;
              int cpuNum = pbsNodeStatus.getCPUNum();
//System.err.println("**** Reduce CPU num by half for 2-thread jobs ****");
//int cpuNum = pbsNodeStatus.getCPUNum()/2;
              Collection<String> threadJobIDs = pbsNodeStatus.getRunningJobIDs();  
              Collection<String> jobIDs = new HashSet(threadJobIDs);
              int jobNum = 0;
              if (jobIDs != null) {
                 jobNum = jobIDs.size();
              }

              int now = (int) (Calendar.getInstance().getTimeInMillis()/1000);

              /*
               * Calculate available time of all busy CPUs
               */
              if (jobNum > 0) {
                 for (String jobID : jobIDs) {
                    PBSJobStatus pbsJobStatus = (PBSJobStatus) statusMgr.getJob(jobID);
                    UniformCPU cpu = new UniformCPU(pbsNodeStatus.getName() + "/" + cpuCount);
                    cpu.setNode(pbsNodeStatus.getName());
                    String pfi = pbsNodeStatus.getAttribute("pfi");
                    if (pfi != null) {
                       cpu.setPerformanceIndex(Double.parseDouble(pfi));
                    }
//                    int remainingTime =  pbsJobStatus.getEstimatedExecutionTime() + CLEANUP_TIME - pbsJobStatus.getUsedCPUTime();
                    int usedTime = now - (int) (pbsJobStatus.getExecStartTime().getTimeInMillis()/1000);
                    int remainingTime = (int) ((pbsJobStatus.getEstimatedExecutionTime()/cpu.performanceIndex) - usedTime) + CLEANUP_TIME;
                    if (remainingTime < 0) {
                       remainingTime = Integer.MAX_VALUE;
                       System.err.println("Job " + pbsJobStatus.getName() + " exceeds allowable execution time. Check job validator");
                    }
                    cpu.setAvailableTime(now+remainingTime);
                    cpuCount++;
                    cpus.add(cpu);
                 }
              }

              /*
               * Calculate available time of all idle CPUs
               */
              for (int i=jobNum; i < cpuNum; i++) {
                 UniformCPU cpu = new UniformCPU(pbsNodeStatus.getName() + "/" + cpuCount);
                 cpu.setAvailableTime(now);
                 cpu.setNode(pbsNodeStatus.getName());
                 String pfi = pbsNodeStatus.getAttribute("pfi");
                 if (pfi != null) {
                    cpu.setPerformanceIndex(Double.parseDouble(pfi));
                 }
                 cpuCount++;
                 cpus.add(cpu);
              }
           }

           return cpus;
        }

        public synchronized void setStatusDirtyFlag() {
           dirty_status_flag = true;
        }

        public synchronized void clearStatusDirtyFlag() {
           dirty_status_flag = false;
        }

        public synchronized Schedule getSchedule() {
           return this.schedule;
        }
}
